import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.Writable;
import org.apache.hadoop.io.WritableComparable;
import org.apache.hadoop.mapreduce.Mapper;

import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
import java.util.HashSet;
import java.util.Set;

public class Friend2Map extends Mapper<LongWritable,Text,Text, FriendValue> {
    Set<FriendValue> cleanset = new HashSet<>();
    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        /*思路：
         * key:A时 value有<"B","yes"> 而 KEY:A时value<"C","B">
         * value的第二个值用来表示是否是好友关系，如果不是就用来表示共同好友
         * */
        String line = value.toString();
        String[] clip =  line.split(" ");

        //A B 是文档的输入格式，所以qp数组里的一定是好友关系
        //context.write(new Text(clip[0]),new FriendValue(clip[1],"yes"));
        cleanset.add(new FriendValue(clip[0],clip[1]));
        cleanset.add(new FriendValue(clip[1],clip[0]));
    }

    @Override
    protected void cleanup(Context context) throws IOException, InterruptedException {
        for(FriendValue friendValue : cleanset){
            for (FriendValue friendValue2 : cleanset){
                //这里的相等并不是相等，而是说可以推荐，为了节省脑容量先这么写……
                if(friendValue.equals(friendValue2)){
                    context.write(new Text(friendValue.getFa()),new FriendValue(friendValue2.getFb()," "+friendValue2.getFa()));
                }
            }
        }
    }
}
class FriendValue implements WritableComparable<FriendValue> {
     private String fa , fb; //fb存储是否是好友，yes 如过不是则存储共同好友

    @Override
    public boolean equals(Object obj) {
        FriendValue fv =(FriendValue)obj;
        /*重写思路
        * A B
        * B C
        * 当出现上面这种情况时才认为两个fv “相等”，也就是会存在推荐 可以给A通过B推荐C
        * */
        return this.fb.equals(fv.fa) && !(this.fa.equals(fv.fb));
    }

    public FriendValue() {

     }

     public FriendValue(String a , String b){
         fa = a;
         fb = b;
     }

     public String getFa() {
         return fa;
     }

     public void setFa(String fa) {
         this.fa = fa;
     }

     public String getFb() {
         return fb;
     }

     public void setFb(String fb) {
         this.fb = fb;
     }

     @Override
     public int compareTo(FriendValue o) {
         return 0;
     }

     @Override
     public void write(DataOutput dataOutput) throws IOException {
        dataOutput.writeUTF(this.fa);
        dataOutput.writeUTF(this.fb);
     }

     @Override
     public void readFields(DataInput dataInput) throws IOException {
         this.fa = dataInput.readUTF();
         this.fb = dataInput.readUTF();
     }
 }
